package com.yanxu;

import com.yanxu.domain.Event;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 *  filter 算子的转换。
 *  filter 算子是一种过滤.
 *      我们可以在filter 中对数据进行筛选，从中选出我们需要的元素，丢弃掉那些我们不需要的元素。
 */
public class Api07_Transform_filter {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子
        //从 fromElements 中获取源输入
        DataStreamSource<Event> streamSource = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("zhangsan", "./cart", 2000L),
                new Event("tom", "./likes", 2000L)
        );
        streamSource.print("from elements >>>");

        //使用 filter 进行过滤
        SingleOutputStreamOperator<Event> filterOperator = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event event) throws Exception {
                return event.getUrl().contains("cart");
            }
        });
        //使用 map 进行转换
        SingleOutputStreamOperator<String> mapOperator = filterOperator.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event event) throws Exception {
                return event.getName();
            }
        });
        mapOperator.print("map >>>");

        /*
        * 输出结果：
            from elements >>>> Event(name=Mary, url=./home, duration=1000)
            from elements >>>> Event(name=Bob, url=./cart, duration=2000)
            map >>>> Bob                                                    ----> Bob   是经过filter过滤的
            from elements >>>> Event(name=zhangsan, url=./cart, duration=2000)
            map >>>> zhangsan                                               ----> zhangsan 是经过filter过滤的
            from elements >>>> Event(name=tom, url=./likes, duration=2000)
        * */

        //对环境进行执行。
        env.execute();
    }
}